package com.teamsun.mqtt.handle;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import com.teamsun.mqtt.entity.vo.UserValidate;
import com.teamsun.mqtt.service.ChannelDataService;
import com.teamsun.mqtt.service.UserPwdQueCall;


/**
 * 
 * @author 处理连接handle
 *
 */
@Sharable
public class ConnectionService extends ChannelInboundHandlerAdapter {

	
	/**
	 * 缓存用户登录队列的长度
	 */
	final  int  PRO_LOGIN_SIZE=20;
	
	/**
	 * 用于根据登录的客户端标识找channel
	 */
	ConcurrentHashMap<String, Channel> str2channel;

	/**
	 * 用于根channel 找登录的客户端
	 */
	ConcurrentHashMap<Channel, String> channel2str;
	
	/**
	 * 操作数据库线程组
	 */
	ExecutorService executorService;
	
	/**
	 * 当前登录的连接数量
	 */
	AtomicInteger loginconns;//当前登录的连接数量

	/**
	 * 缓存用于验证用户登录的队列  当队列满的时候去数据库验证  防止重复操作数据库io性能下降
	 */
	BlockingQueue<UserValidate> userPwdValidateQue=new LinkedBlockingQueue<UserValidate>(PRO_LOGIN_SIZE);
	
	public ConnectionService(ExecutorService executorService) {
		super();
		this.str2channel = ChannelDataService.getStr2channel();
		this.channel2str = ChannelDataService.getChannel2str();
		this.executorService=executorService;
		loginconns=new AtomicInteger();
	}
	
	public void channelReadComplete(ChannelHandlerContext ctx) {
		ctx.flush();
		try {
			super.channelReadComplete(ctx);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx)
			throws Exception {
		loginout(ctx);
		super.channelInactive(ctx);

	}
	/**
	 * 读取客户端的消息 实际的业务开始
	 * 当前仅仅支持MQTT协议，如果不是MQTT就关闭连接
	 */
	public void channelRead(ChannelHandlerContext ctx, Object msg) {

		if (msg instanceof MqttMessage) {

			MqttMessage message = (MqttMessage) msg;
			MqttMessageType messageType = message.fixedHeader().messageType();

			switch (messageType) {
			case CONNECT:
				loginconns.incrementAndGet();
				MqttConnectMessage connectMessage = (MqttConnectMessage) message;
				ack(ctx, connectMessage);
				loginconns.decrementAndGet();
				break;
			case PINGREQ:
				ping(ctx);
				break;

			case DISCONNECT:
				loginout(ctx);
				break;
			default:
				ctx.fireChannelRead(msg);
				break;
			}
		}

		else
			ctx.channel().close();
	}
	
	/**
	 * 验证用户连接合法性
	 * 
	 * 首先检查用户是否登录过了，如果登录了就拒绝重登。
	 * 检查当前需要登录验证的链接的数量，如果过多就缓存到队列 队列满了去数据库一次执行
	 * 如果比较少就一个一个验证
	 * @param ctx
	 * @param connectMessage
	 */
	private void ack(final ChannelHandlerContext ctx,
			final MqttConnectMessage connectMessage) {

		
		MqttConnectPayload connectPayload=connectMessage.payload();
	
		String ident = connectPayload.clientIdentifier();
		
		MqttFixedHeader fixedHeader = new MqttFixedHeader(
				MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false,0);
		MqttConnAckVariableHeader connectVariableHeader = null;
		
		
		 if(str2channel.containsKey(ident)){//重复登录		
				connectVariableHeader= new MqttConnAckVariableHeader(
						MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false);
				
				MqttConnAckMessage ackMessage = new MqttConnAckMessage(fixedHeader,
						connectVariableHeader);
				ctx.writeAndFlush(ackMessage);
				return;
		}
		
		 UserValidate userValidate=new UserValidate(
				   connectPayload.clientIdentifier(),
					connectPayload.userName(),connectPayload.password(),ctx.channel());
		 
		 
		 if(loginconns.get()>PRO_LOGIN_SIZE){//如果当前登录的过多就放入队列
			 
			 if(! userPwdValidateQue.offer(userValidate)){
					
					List<UserValidate> userValidates=new ArrayList<UserValidate>();
					userPwdValidateQue.drainTo(userValidates);
					subumitLoginCall(userValidates);
				}
			 
			 
		 }
		 else {  //否则就单个处理
			 
				
					List<UserValidate> userValidates=new ArrayList<UserValidate>();
					if(userPwdValidateQue.size()>0){
						userPwdValidateQue.drainTo(userValidates);
					}
					userValidates.add(userValidate);
					subumitLoginCall(userValidates);
			 
		 }


	}

	
	/**
	 * 心跳ping
	 * @param ctx
	 */
	private void ping(ChannelHandlerContext ctx) {

		MqttFixedHeader fixedHeader = new MqttFixedHeader(
				MqttMessageType.PINGRESP, false, MqttQoS.AT_LEAST_ONCE, false,
				0);

		MqttMessage mqttMessage = new MqttMessage(fixedHeader);
		ctx.channel().writeAndFlush(mqttMessage);
	}
	
	
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		cause.printStackTrace();//异常
		ctx.fireExceptionCaught(cause);
		loginout(ctx);

	}
	
	/**
	 * 提交登录任务
	 * @param userValidates
	 */
	void   subumitLoginCall(List<UserValidate> userValidates){
		
		UserPwdQueCall call=new UserPwdQueCall(userValidates, str2channel, channel2str);
		
		executorService.submit(call);
		call=null;
	}
	
	/**
	 * 合法退出
	 * @param channel
	 */
	void  loginout(ChannelHandlerContext handlerContext){
		
		Channel channel=handlerContext.channel();
		String iden = channel2str.get(channel);

		if(iden!=null){
		str2channel.remove(iden);
		System.out.println(iden + "退出" + str2channel.size());

		channel2str.remove(channel);
		System.out.println(channel + "断开" + channel2str.size());
		}
		
		channel.close();
	}
}
